package cn.com.sparkle.paxos;

import java.io.File;
import java.io.IOException;
import java.util.LinkedList;
import java.util.concurrent.Callable;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.log4j.Logger;

import cn.com.sparkle.paxos.addprocess.AddRequestPackage;
import cn.com.sparkle.paxos.deamon.InstanceExecutor;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.Id;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.InstanceVoteRecord;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.SuccessfulRecord;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.Value;
//import cn.com.sparkle.paxos.model.Id;
//import cn.com.sparkle.paxos.model.Value;
import cn.com.sparkle.paxos.stablestorage.ExecuteLogOperator;
import cn.com.sparkle.paxos.stablestorage.ReadSuccessRecordCallback;
import cn.com.sparkle.paxos.stablestorage.RecordFileOperator;
//import cn.com.sparkle.paxos.stablestorage.SuccessRecordFileOperator;
//import cn.com.sparkle.paxos.stablestorage.VoteLogOperator;
import cn.com.sparkle.paxos.stablestorage.event.PrepareRecordRealWriteEvent;
//import cn.com.sparkle.paxos.stablestorage.model.InstancePrepareRecord;
//import cn.com.sparkle.paxos.stablestorage.model.InstanceVoteRecord;
//import cn.com.sparkle.paxos.stablestorage.model.InstanceVotedRecord;
//import cn.com.sparkle.paxos.stablestorage.model.SuccessfulRecord;
import cn.com.sparkle.paxos.util.IdComparator;

public class AccountBook {
	private final static Logger logger = Logger.getLogger(AccountBook.class);

	private final ReentrantLock writeLock = new ReentrantLock();
	private ExecuteLogOperator executeLogOperator;
	// private SuccessRecordFileOperator successRecordFileOperator;
	// private VoteLogOperator voteLogOperator;
	private RecordFileOperator fileOperator;
	private long lastWaitExecuteInstanceId;
	private Configuration configuration;

	public AccountBook(Configuration configuration) throws IOException,
			ClassNotFoundException {
		this(configuration,null);

	}
	
	public AccountBook(Configuration configuration,Long lastExceptInstanceId) throws IOException{
		this.configuration = configuration;
		this.configuration = configuration;
		String dir = configuration.getStableStorage() + "/"
				+ configuration.getSelfAddress().replaceAll(":", "-");

		File executelogDir = new File(dir + "/executelog");

		executeLogOperator = new ExecuteLogOperator(executelogDir);
		lastWaitExecuteInstanceId = executeLogOperator.init();
		if(lastExceptInstanceId != null){
			lastWaitExecuteInstanceId = lastExceptInstanceId;
		}
		logger.info("last waited execute instanceId:"
				+ lastWaitExecuteInstanceId);
	}

	public void initInstanceExecutor(InstanceExecutor instanceExecutor)
			throws ClassNotFoundException, IOException {
		String dir = configuration.getStableStorage() + "/"
				+ configuration.getSelfAddress().replaceAll(":", "-");
		File logDir = new File(dir + "/log");
		fileOperator = new RecordFileOperator(logDir,
				lastWaitExecuteInstanceId, instanceExecutor);
		fileOperator.init();
		// successRecordFileOperator = new SuccessRecordFileOperator(
		// successlogDir, lastWaitExecuteInstanceId, instanceExecutor);
		// long lastCanExecuteInstanceId = successRecordFileOperator.init();
		// voteLogOperator = new VoteLogOperator(votelogDir,
		// lastCanExecuteInstanceId);
	}

	public void finishCurInstance(final long instanceId) throws IOException {
		executeLogOperator.writeExecuteLog(instanceId, new Callable<Object>() {
			@Override
			public Object call() throws Exception {
				fileOperator.finishInstance(instanceId);
				return null;
			}
		});
	}

	public void writeSuccessfulRecord(final long instanceId,
			final SuccessfulRecord successfulRecord,
			LinkedList<AddRequestPackage> addRequestPackages)
			throws IOException {
		final long now = System.currentTimeMillis();
		fileOperator.writeSuccessfulRecord(instanceId, successfulRecord,
				addRequestPackages, null);
	}

	public boolean isSuccessful(long instanceId) {
		return fileOperator.isSuccessful(instanceId);
	}

	/**
	 * 
	 * @param prepareRecord
	 * @param realWriteEvent
	 * @return long -1 is success,other is increaseId conflicted.
	 * @throws IOException
	 */
	public long writePrepareRecord(long instanceId, Id highestJoinNum,
			PrepareRecordRealWriteEvent realWriteEvent) throws IOException {
		try {
			writeLock.lock();
			InstanceVoteRecord.Builder prepareRecordBuilder = InstanceVoteRecord
					.newBuilder().setHighestJoinNum(highestJoinNum);

			if (fileOperator.getVotedInstanceRecordMap()
					.containsKey(instanceId)) {
				InstanceVoteRecord temp = fileOperator
						.getVotedInstanceRecordMap().get(instanceId);
				if (IdComparator.getInstance().compare(
						temp.getHighestJoinNum(),
						prepareRecordBuilder.getHighestJoinNum()) >= 0) {
					return temp.getHighestJoinNum().getIncreaseId();
				} else {
					if (temp.hasHighestVotedNum()) {
						prepareRecordBuilder.setHighestValue(temp
								.getHighestValue());
						prepareRecordBuilder.setHighestVotedNum(temp
								.getHighestVotedNum());
					}

				}
			}
			InstanceVoteRecord r = prepareRecordBuilder.build();
			if(fileOperator.writeVoteRecord(instanceId, r, realWriteEvent)){
				return -1;
			}else{
				return Long.MAX_VALUE -1;
			}
		} finally {
			writeLock.unlock();
		}
	}

	/**
	 * 
	 * @param instanceId
	 * @param highestVotedNum
	 * @param highestValue
	 * @param realWriteEvent
	 * @return long -1 is success,other is increaseId conflicted.
	 * @throws IOException
	 */
	public long writeVotedRecord(long instanceId, Id highestVotedNum,
			Value highestValue, PrepareRecordRealWriteEvent realWriteEvent)
			throws IOException {
		try {
			writeLock.lock();
			if (isSuccessful(instanceId)) {
				return Long.MAX_VALUE;
			}
			InstanceVoteRecord record = InstanceVoteRecord.newBuilder()
					.setHighestJoinNum(highestVotedNum)
					.setHighestVotedNum(highestVotedNum)
					.setHighestValue(highestValue).build();
			InstanceVoteRecord temp = fileOperator.getVotedInstanceRecordMap()
					.get(instanceId);

			if (temp != null
					&& IdComparator.getInstance().compare(
							record.getHighestVotedNum(),
							temp.getHighestJoinNum()) == 0
			// record.getHighestVotedNum().compareTo(
			// temp.getHighestJoinNum()) == 0
			) {
				if(fileOperator
						.writeVoteRecord(instanceId, record, realWriteEvent)){
					return -1;
				}else {
					return Long.MAX_VALUE - 1;
				}
			}
			return temp.getHighestJoinNum().getIncreaseId();
		} finally {
			writeLock.unlock();
		}
	}

	// public SuccessfulRecord getSuccessRecord(long instanceId){
	// try {
	// return successRecordFileOperator.get(instanceId);
	// } catch (Exception e) {
	// logger.error("",e);
	// return null;
	// }
	// }
	public long getMaxInstanceIdInVote() {
		try {
			writeLock.lock();
			return fileOperator.getMaxVoteInstanceId();
		} finally {
			writeLock.unlock();
		}
	}

	public long getLastCanExecutableInstanceId() {
		return fileOperator.getLastExpectSafeInstanceId() - 1;
	}

	public void readSuccessRecord(long fromInstanceId, long toInstanceId,
			ReadSuccessRecordCallback successCallback) throws IOException {
		fileOperator.read(fromInstanceId, toInstanceId, successCallback);
	}

	/**
	 * 
	 * @return -1 indicates there is not remainder in queue
	 */
	public long getFisrtInstnaceIdOfWaitInsertToExecuteQueue() {
		return fileOperator.getFirstInstanceIdInUnsafe();
	}
}
